package process;

import bean.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.junit.Test;

/**
 * @author zhengyonghong
 * @create 2020--12--14--14:33
 */
public class Flink_Process_slideOutput {
    public static class MySlideOutputProcess extends ProcessFunction<String,String>{

        @Override
        public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
            String[] split = value.split(",");
            SensorReading sensorReading = new SensorReading(split[0], Long.parseLong(split[1]), Double.parseDouble(split[2]));
            double temp = Double.parseDouble(split[2]);
            if (temp >= 30.0){
                out.collect(value);
            }else{
                ctx.output(new OutputTag<String>("lower_30"){},value);

            }
        }
    }

    //TODO 用Process做侧输出流，高于30度直接输出，低于30度写到侧输出流
    @Test
    public void test() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        SingleOutputStreamOperator<SensorReading> input = env.readTextFile("input/sensorreading.txt")
                .map((MapFunction<String, SensorReading>) value -> {
                    String[] split = value.split(",");
                    return new SensorReading(split[0], Long.parseLong(split[1]), Double.parseDouble(split[2]));
                })
                .process(new ProcessFunction<SensorReading, SensorReading>() {
                    @Override
                    public void processElement(SensorReading value, Context ctx, Collector<SensorReading> out) throws Exception {
                        if (value.getTemp() > 30.0) {
                            out.collect(value);
                        } else {
                            ctx.output(new OutputTag<SensorReading>("output") {
                            }, value);
                        }
                    }
                });
        input.print("high");
        input.getSideOutput(new OutputTag<SensorReading>("output"){}).print("low");
        env.execute();
    }

}
